package com.caul.demo.hadoop.hdfs.client;

import com.caul.demo.hadoop.hdfs.Cmd;
import com.caul.demo.hadoop.hdfs.model.MetaData;
import com.caul.demo.hadoop.hdfs.model.NameNodeReq;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.Arrays;

/**
 * Created by BlueDream on 2018-03-27.
 */
public class DownloadClient {

  public static void main(String[] args) throws IOException {
    System.out.println("启动下载客户端:-------");

    String fileName = "demo.log";
    File file = new File(Cmd.DATA_PATH + "/download/" + fileName);

    //与namenode通讯,請求上传文件,返回blockid和datanode节点
    MetaData nameNodeResp = getDownloadInfo(fileName);
    //链接datanode,下载文件
    for (MetaData.BlockData blockData : nameNodeResp.getBlocks()) {
      downloadData(file, blockData);
    }
  }

  /**
   * 与namenode通讯,請求上传文件
   * 返回blockid和datanode节点
   *
   * @return
   * @throws IOException
   */
  private static MetaData getDownloadInfo(String fileName) throws IOException {
    Socket nameNode = new Socket(Cmd.NAMENODE_IP, Cmd.NAMENODE_PORT);
    OutputStream nnOut = nameNode.getOutputStream();
    InputStream nnIn = nameNode.getInputStream();

    try {
      nnOut.write(new NameNodeReq(Cmd.DOWNLOAD, fileName, -1).value().getBytes());
      nnOut.flush();

      byte[] bts = new byte[1024];
      int read = nnIn.read(bts);
      if (read == -1) {
        throw new RuntimeException("文件不存在:" + fileName);
      }

      return MetaData.of(new String(bts, 0, read));
    } finally {
      nnIn.close();
      nnOut.close();
      nameNode.close();
    }
  }

  private static void downloadData(File file, MetaData.BlockData blockData) throws IOException {
    Socket dataNode = new Socket(blockData.getDataNode(), Cmd.DATANODE_PORT);
    OutputStream dnOut = dataNode.getOutputStream();
    InputStream dnIn = dataNode.getInputStream();
    FileOutputStream fos = new FileOutputStream(file, true);

    try {
      //告诉服务端,开始下载数据
      dnOut.write(Arrays
          .copyOf(String.format(Cmd.DN_DOWNLOAD_REQ, Cmd.DOWNLOAD, blockData.getBlockId())
              .getBytes(), Cmd.DATANODE_CMD_LEN));
      dnOut.flush();

      //写入该block数据
      byte[] bts = new byte[4096];
      int read = 0;
      while ((read = dnIn.read(bts)) != -1) {
        fos.write(bts, 0, read);
      }
      fos.flush();
    } finally {
      fos.close();
      dnIn.close();
      dnOut.close();
      dataNode.close();
    }
  }

}
